package com.dong.rabbitmq.confirm;

import com.dong.rabbitmq.utils.RabbitMqUtil;
import com.dong.rabbitmq.utils.SleepUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 发布确认消息
 *
 * @author dongjian
 */
public class ConfirmMsg {

    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        //1.单个确认 1000ms
//        messagePublishIndividually(channel, queueName);
        //2.批量确认 108ms
//        messagePublishBatch(channel, queueName);
        //3.异步确认 发布1000条，异步确认所用时间为:59ms
        messagePublishAsync(channel, queueName);
        //4.异步发布确认对发布失败的数据进行记录 发布1000条，异步确认所用时间为:57ms
    }

    /**
     * 发布确认-单个确认
     *
     * @throws Exception
     */
    public static void messagePublishIndividually(Channel channel, String queueName) throws Exception {
        // 记录开始时间
        SleepUtil.begin();
        // 批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = i + "";
            channel.basicPublish("", queueName, null, msg.getBytes("UTF-8"));
            // 单个消息发送后马上进行消息确认
            boolean b = channel.waitForConfirms();
            if (b) {
                System.out.println("消息发送成功：" + i);
            }

        }
        // 计算所用的时间
        long l = SleepUtil.comPutTime();
        System.out.println("发布" + MESSAGE_COUNT + "条，单个确认所用时间为:" + l + "s");
    }

    /**
     * 发布确认-批量确认
     *
     * @param channel   信道
     * @param queueName 队列名称
     * @throws Exception {@link Exception}
     */
    public static void messagePublishBatch(Channel channel, String queueName) throws Exception {
        SleepUtil.begin();
        // 批量单位
        int batchSize = 100;
        // 批量发送消息
        for (int i = 1; i <= MESSAGE_COUNT; i++) {
            String msg = i + "";
            channel.basicPublish("", queueName, null, msg.getBytes("UTF-8"));
            // 如果达到每次批量次数则进行确认
            if (i % batchSize == 0) {
                // 确认方法
                channel.waitForConfirms();
            }
        }
        // 计算所用的时间
        long l = SleepUtil.comPutTime();
        System.out.println("发布" + MESSAGE_COUNT + "条，批量确认所用时间为:" + l + "ms");
    }

    /**
     * 发布确认-异步确认
     * @param channel 信道
     * @param queueName 队列名称
     * @throws Exception
     */
    public static void messagePublishAsync(Channel channel, String queueName) throws Exception {
        ConcurrentSkipListMap<Long, String> confirmMessageMap = new ConcurrentSkipListMap<>();
        /**
         * 参数说明：
         * 1.消息标记
         * 2.是否批量
         */
        // 记录那些消息发送成功了
        ConfirmCallback ackCallBack = (deliveryTag, multiple) -> {
            // 2.发布成功的数据不需要记录，去掉发布成功的记录
            System.out.println("消息发布确认成功" + deliveryTag + ",multiple:"+multiple);
            if (multiple) {
                // 批量删除
                confirmMessageMap.headMap(deliveryTag+1).clear();
            } else {
                // 单个删除
                confirmMessageMap.remove(deliveryTag);
            }
        };
        // 记录那些消息发送失败了
        ConfirmCallback nackCallBack = (deliveryTag, multiple) -> {
            // 打印未确认的消息 todo 重发
            String message = confirmMessageMap.get(deliveryTag);
            System.out.println("未确认的消息是：" + message +"======>消息发布确认失败的消息tag：" + deliveryTag);
        };
        // 添加消息发送监听器，记录消息发送成功和失败,异步的
        channel.addConfirmListener(ackCallBack, nackCallBack);
        SleepUtil.begin();
        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String msg = i + "";
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", queueName, null, msg.getBytes("UTF-8"));
            // 1.记录所有发布的消息
            confirmMessageMap.put(nextPublishSeqNo, msg);
        }
        long l = SleepUtil.comPutTime();
        System.out.println("发布" + MESSAGE_COUNT + "条，异步确认所用时间为:" + l + "ms");
    }
}
